package com.flute.icrawler.app.result;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.flute.haflute.jobbox.base.storage.KeyValueStorageService;
import com.flute.haflute.jobbox.base.storage.StorageServiceFactory;
import com.flute.haflute.monitor.GeneralStatusCenter;
import com.flute.haflute.monitor.StatusMessage;
import com.flute.icrawler.app.entity.CrawlResult;
import com.flute.icrawler.app.service.CrawlService;

/**
 * Cassandra数据库存储方式
 * 
 */
public class CassandraStoreImp implements IStoreable {

	private static final Logger LOGGER = LoggerFactory
			.getLogger(CassandraStoreImp.class);

	private KeyValueStorageService service = null;
	private KeyValueStorageService timeService = null;

	public CassandraStoreImp() {
		// TODO Auto-generated constructor stub
		try {
			service = StorageServiceFactory.getStorageService();
			timeService = StorageServiceFactory.getStorageService(true);

		} catch (Exception e) {
			// TODO Auto-generated catch block
			LOGGER.error("CassandraStoreImp:{}", e);
			StatusMessage statusMessage = new StatusMessage();
			statusMessage.topic = "数据库初始化失败";
			statusMessage.content = "数据库初始化失败";
			LOGGER.error("CassandraStoreImp:{}", "Cassandra数据库初始化失败");
			GeneralStatusCenter.getInstance().report(statusMessage);
		}
	}

	@Override
	public String store(CrawlResult crawlResult) {

		try {

			if (null == service || null == timeService) {
				//
				String msg = "链接数据库状态:" + (null == service) + "时间戳数据库状态:"
						+ (null == timeService) + crawlResult.getUrl() + "写入失败";

				StatusMessage statusMessage = new StatusMessage();
				statusMessage.topic = "数据库异常";
				statusMessage.content = msg;

				LOGGER.error("CassandraStoreImp:{}", msg);
				GeneralStatusCenter.getInstance().report(statusMessage);
				return null;
				//
			}

			String time = "" + System.currentTimeMillis();
			if (crawlResult.getContentType().contains("image")) {
				timeService.insert(time, crawlResult.getUrl());
			} else {
				service.insert(time + "_" + crawlResult.getUrl(), new String(
						crawlResult.getContent(), crawlResult.getCharSet()));
				timeService.insert(time, crawlResult.getUrl());
			}

		} catch (Exception e) {
			LOGGER.error("CassandraStoreImp:", e);
			String msg = "Cassandra err=" + crawlResult.getUrl();
			CrawlService.getInstance()
					.logResult(crawlResult.getHostName(), msg);
			return null;
		}

		LOGGER.info("CassandraStoreImp:{}", "save ok:" + crawlResult.getUrl()
				+ "->");
		return "";

	}

	@Override
	public void close() {
		// TODO Auto-generated method stub
		// StorageServiceFactory.release(service);
	}

	public static void main(String[] args) {
		CassandraStoreImp cassandraStoreImp = new CassandraStoreImp();
		System.out.println(cassandraStoreImp);
	}

}
